package cn.hfuu.flink;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

/**
 * @author oldliu
 * @since 1.0
 */
public class WatserSensorWindow5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //env.setParallelism(4);
        //connector
        DataStreamSource<String> source = env.socketTextStream("hadoopmaster", 7777);
        KeyedStream<WaterSensor, String> window = source.map(line -> {
            String[] split = line.split(",");
            if (split.length == 3) {
                return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
            }
            return null;
        }).filter(x -> x != null).keyBy(x -> x.getId());

        //.window(TumblingProcessingTimeWindows.of(Duration.ofMinutes(2)));
//        SingleOutputStreamOperator<WaterSensor> r = window.reduce((x, y) -> {
//            return new WaterSensor(x.getId(), Math.max(x.getTs(), y.getTs()), x.getVc() + y.getVc());
//        });

        SingleOutputStreamOperator<WaterSensor> max =
                window.window(TumblingProcessingTimeWindows.of(Duration.ofMinutes(2))).maxBy("vc");
        System.out.println("最大水位：");
        max.print();
        env.execute();
    }
}
